package com.ibm.cps.sql.window;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.sql.ProcessorMetadataCreator;
import com.ibm.cps.sql.expression.ArrayVisitingNode;
import com.ibm.cps.sql.expression.BooleanExpression;
import com.ibm.cps.sql.expression.CalculationNode;
import com.ibm.cps.sql.expression.ComparisonExpression;
import com.ibm.cps.sql.expression.ConditionCompositionExpression;
import com.ibm.cps.sql.expression.ConditionCompositionExpression.Predicate;
import com.ibm.cps.sql.expression.ExpressionNode;
import com.ibm.cps.sql.expression.FunctionNode;
import com.ibm.cps.sql.expression.IdentifierNode;
import com.ibm.cps.sql.expression.NumberNode;
import com.ibm.cps.sql.expression.StringLiteralNode;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.expression.function.FunctionRegister;
import com.ibm.cps.sql.expression.function.meta.FunctionMetaCreator;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class WindowProcessorCreator {
	private final ProcessorMetadataCreator metaCreator;
	private final String groupId;
	private final String ruleName;
	private int temprorayIndex = 0;

	public WindowProcessorCreator(String tenantId, String groupId, String ruleName) {
		this.metaCreator = new ProcessorMetadataCreator(tenantId, groupId, "sql");
		this.groupId = groupId;
		this.ruleName = ruleName;
	}
	
	public ArrayNode createProcessorsForCondition(String parentProcessorId, BooleanExpression filteringCondition) throws CPSException{
		ArrayNode result = JSONUtility.newArrayNode();
		SimplifiedCondition simplified = simplify(parentProcessorId, filteringCondition);
		result = simplified.processors;
		ObjectNode simplifiedRuleNode = metaCreator.createNaiveRuleNode(simplified.lastProcessorId,
				false, simplified.condition, ruleName);
		result.add(simplifiedRuleNode);
		return result;
	}
	
	private SimplifiedCondition simplify(String parentProcessorId, BooleanExpression condition) throws CPSException{
		if(condition instanceof ConditionCompositionExpression){
			ConditionCompositionExpression expression = (ConditionCompositionExpression)condition;
			BooleanExpression condition1 = expression.getCondition1();
			BooleanExpression condition2 = expression.getCondition2();
			Predicate predicate = expression.getPredicate();
			SimplifiedCondition simplified1 = simplify(parentProcessorId, condition1);
			parentProcessorId = simplified1.lastProcessorId;
			ArrayNode processors = simplified1.processors;
			SimplifiedCondition simplified2 = simplify(parentProcessorId, condition2);
			processors.addAll(simplified2.processors);
			BooleanExpression simplifiedCondition = new ConditionCompositionExpression(simplified1.condition, simplified2.condition, predicate);
			return new SimplifiedCondition(simplifiedCondition, processors, simplified2.lastProcessorId);
		}else{
			ArrayNode processors = JSONUtility.newArrayNode();
			ComparisonExpression expression = (ComparisonExpression)condition;
			String[] parentId = new String[]{parentProcessorId};
			ValueExpression left = expression.getLeftOperand();
			String leftTempName = "__condition__left__operand__" + temprorayIndex;
			temprorayIndex++;
			ValueExpression simplifiedLeft = simplifyConditionOperand(parentId, left, leftTempName, processors);
			ValueExpression right = expression.getRightOperand();
			String rightTempName = "__condition__right__operand__" + temprorayIndex;
			temprorayIndex++;
			ValueExpression simplifiedRight = simplifyConditionOperand(parentId, right, rightTempName, processors);
			BooleanExpression simplifiedCondition = new ComparisonExpression(expression.getOperator(), simplifiedLeft, simplifiedRight);
			return new SimplifiedCondition(simplifiedCondition, processors, parentId[0]);
		}
	}
	
	private ValueExpression simplifyConditionOperand(String[] parentProcessorId, ValueExpression operand, String tempName, ArrayNode processors) throws CPSException{
		String parentId = parentProcessorId[0];
		if (operand instanceof IdentifierNode || operand instanceof ArrayVisitingNode || operand instanceof NumberNode || operand instanceof StringLiteralNode){
			return operand;
		}else{
			ProcessorDescriptor expanding = create(parentId, operand, tempName);
			parentProcessorId[0] = expanding.lastProcessorId;
			processors.addAll(expanding.processors);
			IdentifierNode simplifiedNode = new IdentifierNode(expanding.outputFieldName);
			return simplifiedNode;
		}
	}
	
	private static class SimplifiedCondition{
		private final BooleanExpression condition;
		private final ArrayNode processors;
		private final String lastProcessorId;

		public SimplifiedCondition(BooleanExpression condition,
				ArrayNode processors, String lastProcessorId) {
			this.condition = condition;
			this.processors = processors;
			this.lastProcessorId = lastProcessorId;
		}
	}

	public ProcessorDescriptor create(String parentProcessorId, ExpressionNode expression, String outputFieldName) throws CPSException {
		if (expression instanceof CalculationNode) {
			return createProcessorsFromCalculationNode(parentProcessorId, (CalculationNode) expression, outputFieldName);
		} else if (expression instanceof FunctionNode) {
			return createProcessorsFromFunctionNode(parentProcessorId, (FunctionNode) expression, outputFieldName);
		} else if (expression instanceof IdentifierNode || expression instanceof ArrayVisitingNode || expression instanceof NumberNode || expression instanceof StringLiteralNode){
			ArrayNode emptyNode = JSONUtility.newArrayNode();
			return new ProcessorDescriptor(emptyNode, null, null, expression);
		} else{
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "Unsupported expression type " + expression.getClass());
		}
	}

	private ProcessorDescriptor createProcessorsFromCalculationNode(String parentProcessorId, CalculationNode node,
			String outputFieldName) throws CPSException {
		ArrayNode result = JSONUtility.newArrayNode();
		ExpressionNode left = node.getLeftOperand();
		ExpressionNode right = node.getRightOperand();
		String leftResultFieldName = "__tmp_field_calculation_operand_left_" + (temprorayIndex++);
		String rightResultFieldName = "__tmp_field_calculation_operand_right_" + (temprorayIndex++);
		// Calculation node is responsible of removing temproray fields created
		// by operands
		ProcessorDescriptor leftOutput = create(parentProcessorId, left, leftResultFieldName);
		if (leftOutput.processors.size() != 0)
			parentProcessorId = leftOutput.lastProcessorId;
		ProcessorDescriptor rightOutput = create(parentProcessorId, right, rightResultFieldName);
		if (rightOutput.processors.size() != 0)
			parentProcessorId = rightOutput.lastProcessorId;
		temprorayIndex++;
		
		ObjectNode windowCalculationNode = metaCreator.createVectorCalculationNode(parentProcessorId, node.getOperator().toString(), leftOutput.outputExpression, rightOutput.outputExpression, outputFieldName);
		String calculationNodeId = windowCalculationNode.get(MessageFields.PROCESSORID).asText();
		result.addAll(leftOutput.processors);
		result.addAll(rightOutput.processors);
		result.add(windowCalculationNode);

		IdentifierNode outputExpression = new IdentifierNode(outputFieldName);
		return new ProcessorDescriptor(result, calculationNodeId, outputFieldName, outputExpression);
	}

	private ProcessorDescriptor createProcessorsFromFunctionNode(String parentProcessorId, FunctionNode node,
			String outputFieldName) throws CPSException {
		ArrayNode result = JSONUtility.newArrayNode();
		ExpressionNode[] argumentExpressions = new ExpressionNode[node.getArguments().length];
		String[] argTypes = new String[node.getArguments().length];
		for (int i = 0; i < node.getArguments().length; i++) {
			ExpressionNode arg = node.getArguments()[i];
			String tempOutputFielName = "__tmp_field_function_argument_" + (temprorayIndex++);
			ProcessorDescriptor processorsForArg = create(parentProcessorId, arg, tempOutputFielName);
			if (processorsForArg.processors.size() != 0) {
				parentProcessorId = processorsForArg.lastProcessorId;
			}
			result.addAll(processorsForArg.processors);
			ExpressionNode outputExpression = processorsForArg.outputExpression;
			argumentExpressions[i] = outputExpression;
			argTypes[i] = outputExpression.getClass().getName();
		}
		String functionName = node.getFunctionName();
		Class<?> processorClass = FunctionRegister.findVectorFunction(functionName);
		FunctionMetaCreator creator = new FunctionMetaCreator(processorClass);
		ObjectNode functionNode = creator.create(functionName, parentProcessorId, groupId, outputFieldName, argumentExpressions);
		result.add(functionNode);
		String functionNodeId = functionNode.get(MessageFields.PROCESSORID).asText();
		IdentifierNode outputExpression = new IdentifierNode(outputFieldName);
		return new ProcessorDescriptor(result, functionNodeId, outputFieldName, outputExpression);
	}

	public static class ProcessorDescriptor {
		public final ArrayNode processors;
		public final String lastProcessorId;
		public final String outputFieldName;
		public final ExpressionNode outputExpression;

		public ProcessorDescriptor(ArrayNode processors, String processorIdofLastProcessor,
				String outputFieldName, ExpressionNode outputExpression) {
			this.processors = processors;
			this.lastProcessorId = processorIdofLastProcessor;
			this.outputFieldName = outputFieldName;
			this.outputExpression = outputExpression;
		}

		@Override
		public String toString() {
			return "{\"ProcessorDescriptor\":{\"processors\":\"" + processors
					+ "\", \"processorIdOfLastProcessor\":\""
					+ lastProcessorId + "\", \"outputFieldName\":\""
					+ outputFieldName + "\", \"outputExpression\":\""
					+ outputExpression + "\"}}";
		}
	}
}
